package hit.edu.zjc.Operator;


import hit.edu.zjc.Tool.ObjectConvert;
import hit.edu.zjc.Tool.RWObject;
import hit.edu.zjc.Tool.ShowOutPut;
import hit.edu.zjc.marshalling.ByteSerializerDeserializer;
import hit.edu.zjc.parsers.ByteTupleParserFactory;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.IOperatorDescriptorRegistry;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;

public class SourceStringListIntegrationOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
    private static final long serialVersionUID = 1L;
    private String[] FilePaths;//含有HashSet的二进制的文件路径
    public SourceStringListIntegrationOperatorDescriptor(IOperatorDescriptorRegistry spec,String[] files) {
        super(spec, 0, 1);  
        recordDescriptors[0]  = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});
        FilePaths=files;
    }

    @Override
    public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
    	 final ITupleParser parser =new ByteTupleParserFactory().createTupleParser(ctx);
    	return new AbstractUnaryOutputSourceOperatorNodePushable() {

    		//将otherlist合并到stringlist上。
    			public void Merge(ArrayList<String[]>StringList,ArrayList<String[]>OtherStringList)
    			{
    				StringList.addAll(OtherStringList);
    			}
    		
            @Override
            public void initialize() throws HyracksDataException {

            	final int filelength=FilePaths.length;
            	ArrayList<String[]> StringList=(ArrayList<String[]>)RWObject.ReadObject(FilePaths[0]);
            	//把其他List合并到第一个上。
            	for(int i=1;i<filelength;i++)
            	{
            		ArrayList<String[]> OtherStringList=(ArrayList<String[]>)RWObject.ReadObject(FilePaths[i]);
            		Merge(StringList,OtherStringList);
            	}
            	//合并结束。
      	
             writer.open();
             try
             	{
	     	    	byte[] SetListBytes =ObjectConvert.ObjectToByteArray(StringList);  		   
	     		   ByteArrayInputStream bais = new ByteArrayInputStream(SetListBytes);	
	     	    	parser.parse(bais, writer);
	     	      bais.close();
	            
             	}
             catch(Exception e)
             	{
            	 	writer.fail();
            	 	e.printStackTrace();
             	}
             finally
             	{
            	 	writer.close();
             	}
            }
        };
    }
}